/*
    asyframe for asyframe@gmail.com

    Copyright (C) 2011, asyframe@gmail.com, http://asyframe.googlecode.com/

    This program is free software: you can redistribute it and/or modify
    it under the terms of the GNU General Public License as published by
    the Free Software Foundation, either version 3 of the License, or
	(at your option) any later version.

    This program is distributed in the hope that it will be useful,
    but WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
    GNU General Public License for more details.

    You should have received a copy of the GNU General Public License
	along with this program.  If not, see <http://www.gnu.org/licenses/>.
*/

#include "stdafx.h"
#include "ConstDef.h"
#include "Common/Logger.h"
#include "Common/Interface.h"
#include "Common/ThreadHandler.h"
#include "Common/WorkerBase.h"
#include "Common/AppBase.h"
#include "Common/BufferPool.h"
#include "Common/IoBuffer.h"
#include "Network/DataHandler.h"

namespace asyframe
{
	DataHandlerContext::DataHandlerContext()
		:pBuffer_(),bytesToTransfer_(0)
	{

	}

	DataHandlerContext::DataHandlerContext( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext )
		:HandlerContextBase(_pDataHandlerContext),pBuffer_(),bytesToTransfer_(_pDataHandlerContext->bytesToTransfer_)
	{

	}

	DataHandlerContext::~DataHandlerContext()
	{
		pBuffer_.reset();
		bytesToTransfer_ = 0;
	}

///////////////////////////////////////////////////////////////////////////////////////////

	UdpHandlerContext::UdpHandlerContext()
		:pRemoteEndpoint_()
	{

	}

	UdpHandlerContext::UdpHandlerContext( boost::shared_ptr<UdpHandlerContext> _pUdpHandlerContext )
		:DataHandlerContext(_pUdpHandlerContext),pRemoteEndpoint_()
	{
		if(_pUdpHandlerContext->pRemoteEndpoint_)
			pRemoteEndpoint_.reset(new boost::asio::ip::udp::endpoint(*_pUdpHandlerContext->pRemoteEndpoint_));
	}

	UdpHandlerContext::~UdpHandlerContext()
	{
		pRemoteEndpoint_.reset();
	}

///////////////////////////////////////////////////////////////////////////////

	IDataAction::IDataAction()
	{

	}

	IDataAction::~IDataAction()
	{

	}

	retT IDataAction::onRead( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pDataHandlerContext && _pDataHandlerContext->pBuffer_)
		{
			R(_pDataHandlerContext->pBuffer_->produce(_bytesTransferred));
		}
		R(handleTimeout(_pDataHandlerContext,e, _timeoutCount));
		RETSUCCESS;
	}

	retT IDataAction::onWrite( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pDataHandlerContext && _pDataHandlerContext->pBuffer_)
		{
			R(_pDataHandlerContext->pBuffer_->consume(_bytesTransferred));
		}
		R(handleTimeout(_pDataHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

	retT IDataAction::onConnect( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		R(handleTimeout(_pDataHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

///////////////////////////////////////////////////////////////////////////////

	ITcpAction::ITcpAction()
	{

	}

	ITcpAction::~ITcpAction()
	{

	}

	retT ITcpAction::onReadSome( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pDataHandlerContext && _pDataHandlerContext->pBuffer_)
		{
			R(_pDataHandlerContext->pBuffer_->produce(_bytesTransferred));
		}
		R(handleTimeout(_pDataHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

	retT ITcpAction::onWriteSome( boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pDataHandlerContext && _pDataHandlerContext->pBuffer_)
		{
			R(_pDataHandlerContext->pBuffer_->consume(_bytesTransferred));
		}
		R(handleTimeout(_pDataHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

////////////////////////////////////////////////////////////////////////////////////

	IUdpAction::IUdpAction()
	{

	}

	IUdpAction::~IUdpAction()
	{

	}

	retT IUdpAction::onReadFrom( boost::shared_ptr<UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pUdpHandlerContext && _pUdpHandlerContext->pBuffer_)
		{
			R(_pUdpHandlerContext->pBuffer_->produce(_bytesTransferred));
		}
		R(handleTimeout(_pUdpHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

	retT IUdpAction::onWriteTo( boost::shared_ptr<UdpHandlerContext> _pUdpHandlerContext, std::size_t _bytesTransferred, const boost::system::error_code& e, std::size_t _timeoutCount )
	{
		if(_pUdpHandlerContext && _pUdpHandlerContext->pBuffer_)
		{
			R(_pUdpHandlerContext->pBuffer_->consume(_bytesTransferred));
		}
		R(handleTimeout(_pUdpHandlerContext, e, _timeoutCount));
		RETSUCCESS;
	}

/////////////////////////////////////////////////////////////////////////////////////////////////////

	DataHandler::DataHandler()
		:HandlerBase(),DataHandlerProtocol_(DATAHANDLER_PROTOCOL_NULL)
	{

	}

	DataHandler::~DataHandler()
	{
		DataHandlerProtocol_ = DATAHANDLER_PROTOCOL_NULL;
	}

	int DataHandler::getDataHandlerProtocol()
	{
		return DataHandlerProtocol_;
	}

///////////////////////////////////////////////////////////////////////////////

	TcpHandler::TcpHandler()
		:DataHandler(),pSocket_(),pLocalEndpoint_()
	{
		DataHandlerProtocol_ = DATAHANDLER_PROTOCOL_TCP;
	}

	TcpHandler::~TcpHandler()
	{
		pSocket_.reset();
		pLocalEndpoint_.reset();
	}

	retT TcpHandler::bind( boost::shared_ptr<WorkerBase> _pWorker )
	{
		if(!_pWorker)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		R(DataHandler::bind(_pWorker));
		pSocket_.reset(new boost::asio::ip::tcp::socket(*_pWorker->getIoService()));
		if(pLocalEndpoint_)
		{
			pSocket_->open(pLocalEndpoint_->protocol());
			pSocket_->set_option(boost::asio::ip::tcp::socket::reuse_address(true));
			pSocket_->bind(*pLocalEndpoint_);
		}
		RETSUCCESS;
	}

	retT TcpHandler::asyncRead( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onRead, pIDataAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->space())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->space();
		boost::asio::async_read(*pSocket_,
			boost::asio::buffer(_pBuffers->blank(),bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT TcpHandler::asyncWrite( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onWrite, pIDataAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->size())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->size();
		boost::asio::async_write(*pSocket_,
			boost::asio::buffer(_pBuffers->data(), bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT TcpHandler::asyncConnect( const void* _pEndpoint, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		R(asyncConnect(*(const boost::asio::ip::tcp::endpoint*)_pEndpoint, _pDataHandlerContext, _dataCallBackFun));
		RETSUCCESS;
	}

	retT TcpHandler::asyncConnect( const std::string& _address, unsigned short _port, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		// Prepare endpoint for connect.
		boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(_address), _port);

		// Connect with the given endpoint.
		R(asyncConnect(endpoint, _pDataHandlerContext, _dataCallBackFun));
		RETSUCCESS;
	}

	retT TcpHandler::asyncConnect( const boost::asio::ip::tcp::endpoint& _endpoint, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onConnect, pIDataAction, _pDataHandlerContext, _1, _2);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, _dataCallBackFun));
		if(!pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		pSocket_->lowest_layer().async_connect(_endpoint,
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	boost::shared_ptr<boost::asio::ip::tcp::socket> TcpHandler::getSocket()
	{
		return pSocket_;
	}

	retT TcpHandler::close()
	{
		R(DataHandler::close());
		if(pSocket_)
		{
			boost::system::error_code ignoredEc;
			pSocket_->lowest_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignoredEc);
			pSocket_->lowest_layer().close();
			pSocket_.reset();
		}
		RETSUCCESS;
	}

	boost::shared_ptr<boost::asio::ip::tcp::endpoint> TcpHandler::getLocalEndpoint()
	{
		return pLocalEndpoint_;
	}

	void TcpHandler::setLocalEndpoint( boost::shared_ptr<boost::asio::ip::tcp::endpoint> _pLocalEndpoint )
	{
		pLocalEndpoint_ = _pLocalEndpoint;
	}

	retT TcpHandler::asyncReadSome( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<ITcpAction> pITcpAction = boost::dynamic_pointer_cast<ITcpAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&ITcpAction::onReadSome, pITcpAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->space())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->space();
		pSocket_->async_read_some(
			boost::asio::buffer(_pBuffers->blank(), bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT TcpHandler::asyncWriteSome( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<ITcpAction> pITcpAction = boost::dynamic_pointer_cast<ITcpAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&ITcpAction::onWriteSome, pITcpAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->size())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->size();
		pSocket_->async_write_some(
			boost::asio::buffer(_pBuffers->data(), bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT TcpHandler::reset()
	{
		if(pSocket_)
		{
			boost::system::error_code ignoredEc;
			pSocket_->lowest_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignoredEc);
			pSocket_->lowest_layer().close();
			pSocket_.reset();
		}
		R(DataHandler::reset());
		RETSUCCESS;
	}

///////////////////////////////////////////////////////////////////////////////

	UdpHandler::UdpHandler()
		:DataHandler(),pSocket_(),pLocalEndpoint_()
	{
		DataHandlerProtocol_ = DATAHANDLER_PROTOCOL_UDP;
	}

	UdpHandler::~UdpHandler()
	{
		pSocket_.reset();
		pLocalEndpoint_.reset();
	}

	retT UdpHandler::bind( boost::shared_ptr<WorkerBase> _pWorker )
	{
		if(!_pWorker)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		R(DataHandler::bind(_pWorker));
		pSocket_.reset(new boost::asio::ip::udp::socket(*_pWorker->getIoService()));
		if(pLocalEndpoint_)
		{
			pSocket_->open(pLocalEndpoint_->protocol());
			pSocket_->set_option(boost::asio::ip::udp::socket::reuse_address(true));
			pSocket_->bind(*pLocalEndpoint_);
		}
		RETSUCCESS;
	}

	retT UdpHandler::asyncRead( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onRead, pIDataAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->space())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->space();
		pSocket_->async_receive(
			boost::asio::buffer(_pBuffers->blank(), bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT UdpHandler::asyncWrite( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onWrite, pIDataAction, _pDataHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pDataHandlerContext)
		{
			if(!_pDataHandlerContext->pBuffer_)
				_pDataHandlerContext->pBuffer_ = _pBuffers;
		}
		std::size_t bytesToTransfer;
		if(_pDataHandlerContext && _pDataHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pDataHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->size())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->size();
		pSocket_->async_send(
			boost::asio::buffer(_pBuffers->data(), bytesToTransfer),
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT UdpHandler::asyncConnect( const void* _pEndpoint, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		R(asyncConnect(*(const boost::asio::ip::udp::endpoint*)_pEndpoint, _pDataHandlerContext, _dataCallBackFun));
		RETSUCCESS;
	}

	retT UdpHandler::asyncConnect( const std::string& _address, unsigned short _port, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		// Prepare endpoint for connect.
		boost::asio::ip::udp::endpoint endpoint(boost::asio::ip::address::from_string(_address), _port);

		// Connect with the given endpoint.
		R(asyncConnect(endpoint, _pDataHandlerContext, _dataCallBackFun));
		RETSUCCESS;
	}

	retT UdpHandler::asyncConnect( const boost::asio::ip::udp::endpoint& _endpoint, boost::shared_ptr<DataHandlerContext> _pDataHandlerContext, BaseCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IDataAction> pIDataAction = boost::dynamic_pointer_cast<IDataAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IDataAction::onConnect, pIDataAction, _pDataHandlerContext, _1, _2);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pDataHandlerContext, _dataCallBackFun));
		if(!pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		pSocket_->lowest_layer().async_connect(_endpoint,
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	boost::shared_ptr<boost::asio::ip::udp::socket> UdpHandler::getSocket()
	{
		return pSocket_;
	}

	retT UdpHandler::close()
	{
		R(DataHandler::close());
		if(pSocket_)
		{
			boost::system::error_code ignoredEc;
			pSocket_->lowest_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignoredEc);
			pSocket_->lowest_layer().close();
			pSocket_.reset();
		}
		RETSUCCESS;
	}

	boost::shared_ptr<boost::asio::ip::udp::endpoint> UdpHandler::getLocalEndpoint()
	{
		return pLocalEndpoint_;
	}

	void UdpHandler::setLocalEndpoint( boost::shared_ptr<boost::asio::ip::udp::endpoint> _pLocalEndpoint )
	{
		pLocalEndpoint_ = _pLocalEndpoint;
	}

	retT UdpHandler::asyncReadFrom( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<boost::asio::ip::udp::endpoint> pRemoteEndpoint_, boost::shared_ptr<UdpHandlerContext> _pUdpHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IUdpAction> pIUdpAction = boost::dynamic_pointer_cast<IUdpAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IUdpAction::onReadFrom, pIUdpAction, _pUdpHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pUdpHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pUdpHandlerContext)
		{
			if(!_pUdpHandlerContext->pBuffer_)
				_pUdpHandlerContext->pBuffer_ = _pBuffers;
			if(!_pUdpHandlerContext->pRemoteEndpoint_)
				_pUdpHandlerContext->pRemoteEndpoint_ = pRemoteEndpoint_;
		}
		std::size_t bytesToTransfer;
		if(_pUdpHandlerContext && _pUdpHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pUdpHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->space())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->space();
		pSocket_->async_receive_from(
			boost::asio::buffer(_pBuffers->blank(), bytesToTransfer),
			*pRemoteEndpoint_,
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT UdpHandler::asyncWriteTo( boost::shared_ptr<IoBuffer> _pBuffers, boost::shared_ptr<boost::asio::ip::udp::endpoint> pRemoteEndpoint_, boost::shared_ptr<UdpHandlerContext> _pUdpHandlerContext, DataCallBackFunType _dataCallBackFun )
	{
		if(_dataCallBackFun == NULL)
		{
			if(boost::shared_ptr<IUdpAction> pIUdpAction = boost::dynamic_pointer_cast<IUdpAction>(pHandlerOwner_.lock()))
			{
				_dataCallBackFun = boost::bind(&IUdpAction::onWriteTo, pIUdpAction, _pUdpHandlerContext, _1, _2, _3);
			}
			else
				RET(ERROR_HANDLERBASE_HANDLEROWNER_NOT_EXIST);
		}
		R(operationStart(_pUdpHandlerContext, boost::bind(_dataCallBackFun, 0, _1, _2)));
		if(!_pBuffers || !pSocket_)
			RET(ERROR_COMMON_SHARED_PTR_NULL);
		if(_pUdpHandlerContext)
		{
			if(!_pUdpHandlerContext->pBuffer_)
				_pUdpHandlerContext->pBuffer_ = _pBuffers;
			if(!_pUdpHandlerContext->pRemoteEndpoint_)
				_pUdpHandlerContext->pRemoteEndpoint_ = pRemoteEndpoint_;
		}
		std::size_t bytesToTransfer;
		if(_pUdpHandlerContext && _pUdpHandlerContext->bytesToTransfer_)
		{
			bytesToTransfer = _pUdpHandlerContext->bytesToTransfer_;
			if(bytesToTransfer > _pBuffers->size())
			{
				LOG_WARN(boost::format( "IoBuffer outof space in %1% %2% %3%" ) % __FUNCTION__ % __FILE__ % __LINE__);
				RET(INFO_DATAHANDLER_IOBUFFER_OUTOF_SPACE);
			}
		}
		else
			bytesToTransfer = _pBuffers->size();
		pSocket_->async_send_to(
			boost::asio::buffer(_pBuffers->data(), bytesToTransfer),
			*pRemoteEndpoint_,
			boost::bind(_dataCallBackFun,
			boost::asio::placeholders::bytes_transferred,
			boost::asio::placeholders::error,
			0));
		RETSUCCESS;
	}

	retT UdpHandler::reset()
	{
		if(pSocket_)
		{
			boost::system::error_code ignoredEc;
			pSocket_->lowest_layer().shutdown(boost::asio::ip::tcp::socket::shutdown_both, ignoredEc);
			pSocket_->lowest_layer().close();
			pSocket_.reset();
		}
		R(DataHandler::reset());
		RETSUCCESS;
	}
} /* namespace asyframe */
